package com.teama4.shop.utils;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.sql.*;
import java.util.List;

public class CommodityRecommendationUtils {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("CommodityRec").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        SparkSession spark = new SparkSession(sc.sc());

        Dataset<Row> jdbcDF = spark.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://192.168.104.101:3306/mall")
                .option("dbtable", "model")
                .option("user", "root")
                .option("password", "123456")
                .load();

        JavaRDD<Row> rawRatings = jdbcDF.javaRDD();

        CommodityRecommendationUtils cru = new CommodityRecommendationUtils();
        JavaRDD<Rating> ratings = cru.buildRating(rawRatings);
        MatrixFactorizationModel model = cru.model(ratings);

        Dataset<Row> user = spark.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://192.168.104.101:3306/mall")
                .option("dbtable", "user")
                .option("user", "root")
                .option("password", "123456")
                .load();

        JavaRDD<Row> u = user.javaRDD();

        List<Integer> list = u.map(line -> {
            Integer i = line.getInt(0);
            return i;
        }).collect();

        for (Integer id : list) {
            Rating[] recommendedCommodity = model.recommendProducts(id, 20);
            String head = "update user set recommended_goods='";
            String tmp = "";
            String commodity = "";
            for(Rating rc : recommendedCommodity) {
                String[] splits = rc.toString().split(",");
                tmp = tmp + splits[1] + ",";
            }
            commodity = head + tmp + "' where id =" + id +";";
            cru.sql(commodity);
        }
    }

    public MatrixFactorizationModel model(JavaRDD<Rating> ratings) {
        return ALS.train(ratings.rdd(), 50, 10, 0.01);
    }

    public JavaRDD<Rating> buildRating(JavaRDD<Row> rawRatings) {
        return rawRatings.map(line -> {
            String[] splits = line.toString().split(",");
            Integer userId = Integer.parseInt(splits[0].substring(1));
            Integer commodityId = Integer.parseInt(splits[1]);
            Integer rating = Integer.parseInt(splits[2].substring(0, splits[2].length()-1));
            return new Rating(userId, commodityId, rating);
        });
    }

    public void recommend(int id) {
        SparkConf conf = new SparkConf().setAppName("CommodityRec").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        SparkSession spark = new SparkSession(sc.sc());

        Dataset<Row> jdbcDF = spark.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://192.168.104.101:3306/mall")
                .option("dbtable", "evaluate")
                .option("user", "root")
                .option("password", "123456")
                .load();

        JavaRDD<Row> rawRatings = jdbcDF.javaRDD();

        CommodityRecommendationUtils cru = new CommodityRecommendationUtils();
        JavaRDD<Rating> ratings = cru.buildRating(rawRatings);
        MatrixFactorizationModel model = cru.model(ratings);
        //以上为model生成

        Rating[] recommendedCommodity = model.recommendProducts(id, 20);
        String head = "update user set recommended_goods='";
        String tmp = "";
        String commodity = "";
        for(Rating rc : recommendedCommodity) {
            String[] splits = rc.toString().split(",");
            tmp = tmp + splits[1] + ",";
        }
        commodity = commodity + head + tmp + "' where id =" + id +";";

        sql(commodity);
    }

    public void sql(String sql) {
        String JDBC_DRIVER = "com.mysql.jdbc.Driver";
        String DB_URL = "jdbc:mysql://localhost:3306/mall";
        String USER = "root";
        String PASS = "123456";

        Connection conn = null;
        Statement stmt = null;
        try{
            // 注册 JDBC 驱动
            Class.forName(JDBC_DRIVER);

            // 打开链接
            System.out.println("连接数据库...");
            conn = DriverManager.getConnection(DB_URL, USER, PASS);

            // 执行查询
            System.out.println(" 实例化Statement对象...");
            stmt = conn.createStatement();
            stmt.executeUpdate(sql);

            // 完成后关闭
            stmt.close();
            conn.close();
        }catch(SQLException se){
            // 处理 JDBC 错误
            se.printStackTrace();
        }catch(Exception e){
            // 处理 Class.forName 错误
            e.printStackTrace();
        }finally{
            // 关闭资源
            try{
                if(stmt!=null) stmt.close();
            }catch(SQLException se2){
            }// 什么都不做
            try{
                if(conn!=null) conn.close();
            }catch(SQLException se){
                se.printStackTrace();
            }
        }
    }
}
